package com.yanduo.tools

import com.yanduo.utils.SchemaUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * p12 日志文件转parquet文件实现方式
  * --采用StructType方式转换 （RDD[Row] + StructType 转换为DataFrame对象，然后写成parquet文件）
  *   采用snappy压缩格式
  *
  * @author Gerry chan
  * @version 1.0
  */
object Bzip2Parquet {
  def main(args: Array[String]): Unit = {
    //0 参数校验
    if (args.length != 3) {
      println(
        """
          |cn.dmp.tools.Bzip2Parquet
          |参数：
          | logInputPath
          | compressionCode <snappy,gzip,lzo>
          | resultOutputPath
          |
        """.stripMargin)
      sys.exit()
    }

    //1 接收程序参数
    val Array(logInputPath, compressionCode, resultOutputPath) = args

    //2 创建sparkconf --> sparkContext

    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")

    //RDD 序列化到磁盘 worker 与 worker之前的数据传输
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    sparkConf.set("spark.sql.parquet.compression.codec", compressionCode)

    // 创建sparkContext
    val sc = new SparkContext(sparkConf)

    val sparkSession = new SparkSession(sc)


    //3 读取日志
    val rawdata = sc.textFile(logInputPath)

    val dataRow: RDD[Row] = rawdata.map(_.split(",", -1))
      .filter(_.length >= 85)
      .map(arr => {
        Row(arr(0), NBF.toInt(arr(1)), NBF.toInt(arr(2)), NBF.toInt(arr(3)))
      })
    //4 根据业务需求对数据进行ETL
    // -1 表示切分到行尾（不管是否有数据）xxx,sx,xxfdf,,,,,,
    // 自定义 NilFormat 反正空字符串转换异常,
    // createDataFrame 需要Row[RDD] 与StructType对象 , 所以先使用map 将数组转换成 RDD[Row] 对象

    //rawdata.map(line => line.split(",", line.length))


    val dataFrame: DataFrame = sparkSession.createDataFrame(dataRow, SchemaUtils.logStructType)
    //5 将结果存储到本地磁盘  createDataFrame 需要Row[RDD] 与StructType对象

    dataFrame.write.parquet(resultOutputPath)
    //6 关闭sc

    sc.stop()

  }
}
